package com.hllxd.hism.star.basic.service.service;


import com.hllxd.hism.star.basic.service.collect.CollectYarn;
import com.hllxd.hism.star.basic.service.collect.OkHttpUtil;
import com.hllxd.hism.star.basic.service.dao.http.RockAssemApiMapper;
import com.hllxd.hism.star.basic.service.dao.mapper.mis.TimeoutTaskMngMapper;
import com.hllxd.hism.star.basic.service.dao.mapper.mis.YarnApplicationMapper;
import com.hllxd.hism.star.basic.service.model.TimeoutTaskEntity;
import com.hllxd.hism.star.basic.service.model.rock.YarnApplicationEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.*;

/**
 * Yarn超时任务管理，进行kill处理
 *
 * @author lixiyong
 * @date 2020/12/20 16:15
 */
@Slf4j
@Service("yarnApplicationService")
public class YarnApplicationServiceImpl implements YarnApplicationService {

    @Autowired
    private TimeoutTaskMngMapper timeoutTaskMngMapper;

    public static final int TIMEOUT_HOURS = 6;
    public static final Set<String> APP_TYPE_SET = new HashSet<>();

    static {
        APP_TYPE_SET.add("MAPREDUCE");
//        APP_TYPE_SET.add("SPARKCORE");
    }


    @Autowired
    YarnApplicationMapper yarnApplicationMapper;

    @Autowired
    RockAssemApiMapper rockAssemApiMapper;

    @Autowired
    CollectYarn collectYarn;


    @Override
    public List<YarnApplicationEntity> listYarnApplication(String ptdate) {
        String ptdate1 = ptdate + " 00:00:00";
        String ptdate2 = ptdate + " 23:59:59";
        return yarnApplicationMapper.listYarnApplication(ptdate1, ptdate2);
    }

    @Override
    public List<YarnApplicationEntity> listYarnApplicationNow(Map<String, String> params) throws IOException {
//        List<YarnQuotaEntity> quotas = collectYarn.listYarnQuota();
//        List<YarnApplicationEntity> applications = collectYarn.listYarnApp();
//        return YarnApplicationResult.build(quotas, applications);
        return CollectYarn.getAppList(params);
    }

    @Override
    public List<YarnApplicationEntity> listYarnApplication4LastTime(String ptdate) {
        log.info("lastTime: {}", ptdate);
        return yarnApplicationMapper.listYarnApplication4LastTime(ptdate);
    }

    @Override
    public YarnApplicationEntity getYarnApplication(Long id) {
        return yarnApplicationMapper.getYarnApplication(id);
    }

//    @Override
//    public Boolean killYarnApplication(String clusterName, String queue, String accessID, String appid) {
//        // 记录对应的操作
//        YarnApplicationKillEntity entity = YarnApplicationKillEntity.build(clusterName, queue, accessID, appid, "KILL-ING");
//        yarnApplicationMapper.insertYarnApplicationKill(entity);
//
//        Boolean ret = rockAssemApiMapper.yarnKillApp(clusterName, appid, accessID);
//        if (ret) {
//            log.info("kill success. {}, {}, {}", clusterName, appid, accessID);
//            yarnApplicationMapper.updateYarnApplicationKillStatus(appid, "KILL-SUCCESS");
//        } else {
//            log.info("kill failed. {}, {}, {}", clusterName, appid, accessID);
//            yarnApplicationMapper.updateYarnApplicationKillStatus(appid, "KILL-FAILED");
//        }
//
//        return true;
//    }

    private static final String KILL_API = "/ws/v1/cluster/apps/%s/state";
    private static final String YARN_URL2 = "http://wq-yarn1-rm2.58dns.org:9088";

    /**
     * @param appId
     * @return
     */
    @Override
    public Boolean killYarnApplication(String appId) {
        OkHttpUtil okHttpUtil = OkHttpUtil.getInstance();
        String api = String.format(KILL_API, appId);
        HashMap<String, String> param = new HashMap<>();
        param.put("state", "KILLED");
        try {
            String request = CollectYarn.YARN_URL + api;
            String res = okHttpUtil.putDataString(request, param);
            if (res.contains("This is standby RM")) {
                request = YARN_URL2 + api;
                res = okHttpUtil.putDataString(request, param);
            }
            log.info("执行killYarnApplication url：{},param:{},result:{}", request, param, res);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return true;
    }

    @Override
    public String checkTimeoutHandler(String instId) throws IOException {
        // todo:: 需要进行优化处理， 不然wjob返回会超时
        //
        log.info("W-JOB, start.timeoutTaskJobHandler, inst: {}.", instId);
        LocalDateTime nowDt = LocalDateTime.now();
        ArrayList<TimeoutTaskEntity> toKillList = new ArrayList<>();
        HashMap<String, String> paramMap = new HashMap<>();
        List<TimeoutTaskEntity> taskEntities = timeoutTaskMngMapper.listTimeoutTask(paramMap);
        Set<String> appSet = new HashSet<>();
        for (TimeoutTaskEntity taskEntity : taskEntities) {
            appSet.add(taskEntity.getAppId());
            Integer optStatus = taskEntity.getOptStatus();
            Long submitTime = taskEntity.getSubmitTime();
            Long delayTime = taskEntity.getDelayTime();
            LocalDateTime submitTimeDt = Instant.ofEpochMilli(submitTime).atZone(ZoneOffset.ofHours(8)).toLocalDateTime();
            long untilHour = submitTimeDt.until(nowDt, ChronoUnit.HOURS);

            if (optStatus == 1) {
                if (untilHour >= TIMEOUT_HOURS) {
                    toKillList.add(taskEntity);
                }
            } else {
                long untilSec = submitTimeDt.until(nowDt, ChronoUnit.SECONDS);
                if (untilSec > TIMEOUT_HOURS * 3600 + delayTime) {
                    toKillList.add(taskEntity);
                }
            }
        }

        Map<String, String> params = new HashMap<>();
        params.put("queue", "");
        List<YarnApplicationEntity> appList = CollectYarn.getAppList(params);
        ArrayList<YarnApplicationEntity> toInsertList = new ArrayList<>();
        for (YarnApplicationEntity yarnApplication : appList) {
            Long startedTime = yarnApplication.getStartedTime();
            String id = yarnApplication.getId();
            String applicationType = yarnApplication.getApplicationType();
            LocalDateTime startedTimeDt = Instant.ofEpochMilli(startedTime).atZone(ZoneOffset.ofHours(8)).toLocalDateTime();
            long until = startedTimeDt.until(nowDt, ChronoUnit.HOURS);
            if (until >= TIMEOUT_HOURS && !appSet.contains(id) && APP_TYPE_SET.contains(applicationType)) {
                toInsertList.add(yarnApplication);
            }
        }

        log.info("toKillList:" + toKillList.size());
        for (TimeoutTaskEntity timeoutTaskEntity : toKillList) {
            log.info("start kill appId:" + timeoutTaskEntity.getAppId());
            killYarnApplication(timeoutTaskEntity.getAppId());
            timeoutTaskMngMapper.updateTimeoutTaskActionByAppId(timeoutTaskEntity.getAppId());
        }

        log.info("toInsertList:" + toInsertList.size());
        for (YarnApplicationEntity yarnApplication : toInsertList) {
            TimeoutTaskEntity entity = new TimeoutTaskEntity();
            entity.setAppId(yarnApplication.getId());
            entity.setAppName(yarnApplication.getName());
            entity.setAppType(yarnApplication.getApplicationType());
            entity.setSubmitTime(yarnApplication.getStartedTime());
            entity.setLaunchTime(yarnApplication.getLaunchTime());
            entity.setUseTime(yarnApplication.getUseTime());
            entity.setQueue(yarnApplication.getQueue());
            entity.setUser(yarnApplication.getUser());
            entity.setOptStatus(1);
            int i = timeoutTaskMngMapper.insert(entity);
        }
        log.info("W-JOB, end.timeoutTaskJobHandler");
        return "SUCCESS";
    }





}
